package com.ict.ms.abnormal.service.rabbitmq.test;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class AbnormalSet {

    private static String IP,user,password,EXCHANGE_NAME;
    private static String EXCHANGE_TYPE = "topic";
    private static boolean EXCHANGE_DURABLE = true;

    private static String QUEUE_NAME;
//    private static int QUEUE_NAME_LENGTH = 15;
    private static boolean QUEUE_DURABLE = false;
    private static boolean QUEUE_EXCLUSIVE = true;
    private static boolean QUEUE_AUTO_DELETE = false;
    private static String bindingKey;
    private static int message_ttl = 60000;    // 单位 ms

    private static Connection connection;
    private static Channel channel;

    /*
    * 若无指定消息过期时间，则默认消息过期时间为60s
    * */
    public AbnormalSet(String IP, String user, String password, String EXCHANGE_NAME, String bindingKey, String QUEUE_NAME){
        this.IP = IP;
        this.user = user;
        this.password = password;
        this.EXCHANGE_NAME = EXCHANGE_NAME;
        this.bindingKey = bindingKey;
        this.QUEUE_NAME=QUEUE_NAME;
    }

    /**
     * 创建连接连接到RabbitMQ
     * @return   0 means success
     *           -1 means failure
     */
    public int BaseConnection(int port){
        ConnectionFactory factory = new ConnectionFactory();
        //设置MabbitMQ所在主机ip或者主机名
        factory.setHost(IP);
        factory.setPort(port);
        factory.setUsername(user);
        factory.setPassword(password);

        //创建一个连接
        try {
            connection = factory.newConnection();
        } catch (IOException e) {
            System.out.println("[y] 请确认输入的IP地址、用户名、密码是否准确！");
            return  -1;
        } catch (Exception e) {
            System.out.println("[y] 连接RabbitMQ超时，请重试！");
            return  -1;
        }

        //创建一个频道
        try {
            channel = connection.createChannel();
        } catch (IOException e) {
            System.out.println("[y] 创建频道出错，请重试！");
            return -1;
        }

        //声明一个路由器，指定名称、模式、及其是否durable
        try {
            channel.exchangeDeclare(EXCHANGE_NAME,EXCHANGE_TYPE,EXCHANGE_DURABLE);
        } catch (IOException e) {
            System.out.println("[y] 路由器声明失败，请重试！");
            return -1;
        }

        //指定一个队列,随机队列名,non-durable,exclusive,not auto-delete
//        QUEUE_NAME = MqConfig.QUEUE_NAME;

        Map<String,Object> args = new HashMap();
        args.put("x-message-ttl",message_ttl);
        try {
            channel.queueDeclare(QUEUE_NAME,QUEUE_DURABLE,QUEUE_EXCLUSIVE,QUEUE_AUTO_DELETE,args);
        } catch (IOException e) {
            System.out.println("[y] 队列声明失败，请重试！");
        }

        try {
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,bindingKey);
        } catch (IOException e) {
            System.out.println("[y] 队列绑定路由关键字 " + bindingKey + " 时出错!");
        }

        System.out.println("[y] 订阅消息者成功连接至RabbitMQ !");
        return 0;
    }

    /**
     *
     * @param type 1：异常设置，2：区域异常添加，3：区域异常删除
     * @return
     */
    public boolean Consume(final int type){
        System.out.println("[y] Receiver " + QUEUE_NAME + " Waiting for messages.To exit press CTRL+C.");
        //消息订阅
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                team.iOceanPlus.PB.Config.PBConfig rec=team.iOceanPlus.PB.Config.PBConfig.parseFrom(body);
                switch (type){
                    case 1:
//                        rec.getConfigAlertAnomaly();
                        System.out.println(rec.getConfigAlertAnomaly());
                        break;
                    case 2:
                        for(int j=0;j<rec.getListConfigZonesOfInterestCount();j++){
                            System.out.println(rec.getListConfigZonesOfInterest(j));
                        }
                        break;
                    case 3:
                        for(int j=0;j<rec.getListConfigTargetsOfInterestCount();j++){
                            System.out.println(rec.getListConfigTargetsOfInterest(j));
                        }
                        break;
                    default:
                        break;
                }
            }
        };

        //消息反馈
        try {
            channel.basicConsume(QUEUE_NAME,false,consumer);
            return true;
        } catch (IOException e) {
            System.out.println("[y] Receiver "+ QUEUE_NAME + " failed to basic consume." );
            return false;
        }
    }

    protected void finalize() throws IOException, Exception {
        channel.close();
        connection.close();
    }
}
